package com.xian.scala.stream

import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.scala._


object MyStreamWC {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val lines: DataStream[String] = env.socketTextStream("localhost", 2323)

    val words: DataStream[String] = lines.flatMap(_.split(" "))

    val word_one: DataStream[(String, Int)] = words.map((_, 1))

    val word_n: KeyedStream[(String, Int), Tuple] = word_one.keyBy(0)

    val word_sum: DataStream[(String, Int)] = word_n.sum(1)

    word_sum.print()


    env.execute()

  }

}
